package com.zyx.flinkdemo.dataset.operator;

import org.apache.flink.api.common.functions.GroupReduceFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.util.*;

/**
 * @author Yaxi.Zhang
 * @since 2021/5/27 10:44
 * desc: GroupReduce案例
 */
public class GroupReduceTemplateTest {
    final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

    DataSet<Tuple2<Integer, String>> dataSet;

    @Before
    public void before() {
        List<Tuple2<Integer, String>> list = new ArrayList<>();

        list.add(new Tuple2<>(15, "classA"));
        list.add(new Tuple2<>(15, "classA"));
        list.add(new Tuple2<>(15, "classB"));
        list.add(new Tuple2<>(23, "classB"));
        list.add(new Tuple2<>(23, "classC"));
        list.add(new Tuple2<>(23, "classC"));
        list.add(new Tuple2<>(44, "classA"));
        list.add(new Tuple2<>(44, "classB"));

        dataSet = env.fromCollection(list);
    }

    @After
    public void after() throws Exception {
        env.execute();
    }

    /**
     * 对同一数字对应的字符串进行去重
     */
    @Test
    public void reduceGroup() throws Exception {
        dataSet
                .groupBy("f0")
                .reduceGroup(new GroupReduceFunction<Tuple2<Integer, String>, Tuple2<Integer, String>>() {
                    final Set<String> distincedWords = new HashSet<>();
                    private Integer key;
                    @Override
                    public void reduce(Iterable<Tuple2<Integer, String>> values,
                                       Collector<Tuple2<Integer, String>> out) throws Exception {
                        for (Tuple2<Integer, String> value : values) {
                            String word = value.f1;
                            key = value.f0;
                            distincedWords.add(word);
                            distincedWords.add(word);
                        }
                        for (String word : distincedWords) {
                            out.collect(Tuple2.of(key, word));
                        }
                    }
                })
                .printOnTaskManager("输出数据");
    }
}
